package com.gis507.test.AisTest3;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.junit.Test;

import java.io.File;
import java.io.IOException;
import java.sql.SQLOutput;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Stream;

/**
 * 1、继承Reducer
 * 2、声明输入输出类型
 * 3、重写reduce方法
 */
@Slf4j
public class AisReducer extends Reducer<Text,AisBean,Text,AisBean> {

    //设置时间阈值
    private  static final long SAMPLING = 10;

    //设置距离阈值
    private static final double STANDTEDISTANCE = 100;

    //设置速度阈值
    private static final double FILTERSPEED = 10;

    //设置最小速度
    private static final double STOPSPEED = 1;

    @Override
    protected void reduce(Text key, Iterable<AisBean> values, Context context) throws IOException, InterruptedException {

        //1、遍历values，将对象封装到列表中
        ArrayList<AisBean> aisList = new ArrayList<>();

        for (AisBean value : values) {

            //将取出的每一个value，放到aisList中
            aisList.add(value);
        }

        //2、对aisList进行排序
        Stream<AisBean> aisListSorted = aisList.stream().sorted((a1, a2) -> Long.compare(a1.getUnixTime(), a2.getUnixTime()));

        //3、对时间超限的删除

    }


    /**
     * 对数组list进行处理
     * @param list
     * @return
     */
    public ArrayList<AisBean> measureTime(ArrayList<AisBean> list){

        for(int i = 1;i < list.size() - 1;i++){

            //i点与i+1点的时间间隔大于等于SAMPLING
            if ((list.get(i + 1).getUnixTime() - list.get(i).getUnixTime()) >= SAMPLING ){

                //i点的速度小于等于FILTERSPEED，然后继续比较和STOPSPEED的大小
                if (list.get(i).getSpeed() <= FILTERSPEED){

                    //比较和STOPSPEED的大小
                    if (list.get(i).getSpeed() <= STOPSPEED){

                        //移除该点
                        list.remove(i);

                    }else {
                        //否则i++，然后打印日志
                        logPrint(list.get(i));
                    }

                }else{
                    //i点的速度大于FILTERSPEED，移除该点
                    list.remove(i);
                }
            }else{

                //i点与i+1点的时间间隔小于SAMPLING，i++继续下一个比较，打印日志
                logPrint(list.get(i));

            }
        }

        //返回list
        return list;

    }

    /**
     * 日志输出
     */
    public void logPrint(AisBean aisBean){

        log.info("内容上传完毕",aisBean);

    }
}
